package cn.me.alphamq.broker.rebalance.strategy.impl;

import cn.me.alphamq.broker.cache.Container;
import cn.me.alphamq.common.model.broker.CommitLogTopicModel;
import cn.me.alphamq.broker.rebalance.ConsumerInstance;
import cn.me.alphamq.broker.rebalance.strategy.RebalanceStrategy;

import java.util.Collections;
import java.util.List;

/**
 * 轮询重平衡策略：打乱消费者集合，然后将队列均分给各个消费者，剩余的队列分给集合前面的消费者
 *
 * @author f
 */
public class RoundRobinRebalance implements RebalanceStrategy {
    @Override
    public void rebalance(String topic, String consumerGroup) {
        CommitLogTopicModel commitLogTopicModel = Container.fetchCommitLogTopicModelMap().get(topic);
        // 队列数
        int queueNumber = commitLogTopicModel.getQueueList().size();

        List<ConsumerInstance> consumerGroupInstanceList = Container.getConsumerInstanceMap()
                .get(topic)
                .get(consumerGroup);
        // 消费者数
        int consumerNumber = consumerGroupInstanceList.size();

        // 恢复消费者实例 queueIdList 的空集合
        for (ConsumerInstance consumerInstance : consumerGroupInstanceList) {
            consumerInstance.restoreQueueIdList();
        }

        // 打乱消费者实例的顺序，确保消费者消费公平
        Collections.shuffle(consumerGroupInstanceList);
        for (int i = 0; i < queueNumber; i++) {
            consumerGroupInstanceList.get(i % consumerNumber).getQueueIdList().add(i);
        }
    }
}
